package com.atguigu.gmall.realtime.app.function;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.text.ParseException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author jason
 * @create 2022-04-16 21:54
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimAsyncJoinFunction<T> {
    private Connection connection;
    private ThreadPoolExecutor threadPoolExecutor;
    private String tableName;

//    //    不能确定的时候 当抽象方法 外部调用
//    public abstract String getKey(T input);
//
//    //    不能确定的时候 当抽象方法 外部调用
//    public abstract void join(T input, JSONObject dimInfo) throws ParseException;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);

        threadPoolExecutor = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
//              获取查询的主键
                    String id = getKey(input);

//              查询维度信息
                    JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, id);

//              补充维度信息
                    if (dimInfo != null) {
                        join(input, dimInfo);
                    }

//              将数据输出
                    resultFuture.complete(Collections.singletonList(input));

                } catch (Exception e) {
                    e.printStackTrace();
                }


            }


        });
//        发送请求

//        合并数据

//        外面调用

    }


    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("TimeOut ：" + input);
//        生产环境如果超时可以在这里再发一次请求
    }
}
